//! Flight SQL服务器实现

use std::net::SocketAddr;
use std::sync::Arc;
use tonic::transport::Server;
use arrow_flight::flight_service_server::FlightServiceServer;
use data_gateway_kernel::{DataGatewayKernel, Result};
use crate::service_simple::SimpleFlightSqlService;

/// Flight SQL服务器配置
#[derive(Debug, Clone)]
pub struct FlightSqlServerConfig {
    /// 监听地址
    pub bind_address: SocketAddr,
    /// 是否启用TLS
    pub enable_tls: bool,
    /// TLS证书路径
    pub tls_cert_path: Option<String>,
    /// TLS私钥路径
    pub tls_key_path: Option<String>,
    /// 最大并发连接数
    pub max_concurrent_connections: usize,
    /// 请求超时时间（秒）
    pub request_timeout_seconds: u64,
}

/// Flight SQL服务器
pub struct FlightSqlServer {
    /// 配置
    config: FlightSqlServerConfig,
    /// 内核引用
    kernel: Arc<DataGatewayKernel>,
    /// 服务实例
    service: Arc<SimpleFlightSqlService>,
}

impl Default for FlightSqlServerConfig {
    fn default() -> Self {
        Self {
            bind_address: "127.0.0.1:50051".parse().unwrap(),
            enable_tls: false,
            tls_cert_path: None,
            tls_key_path: None,
            max_concurrent_connections: 1000,
            request_timeout_seconds: 300, // 5分钟
        }
    }
}

impl FlightSqlServer {
    /// 创建新的Flight SQL服务器
    pub fn new(config: FlightSqlServerConfig, kernel: Arc<DataGatewayKernel>) -> Self {
        let service = Arc::new(SimpleFlightSqlService::new(kernel.clone()));

        Self {
            config,
            kernel,
            service,
        }
    }
    
    /// 启动服务器
    pub async fn start(&self) -> Result<()> {
        tracing::info!("启动Flight SQL服务器: {}", self.config.bind_address);
        
        let service = FlightServiceServer::new((*self.service).clone())
            .max_decoding_message_size(64 * 1024 * 1024) // 64MB
            .max_encoding_message_size(64 * 1024 * 1024); // 64MB
        
        let mut server_builder = Server::builder()
            .concurrency_limit_per_connection(self.config.max_concurrent_connections);
        
        // 暂时禁用TLS配置以简化实现
        if self.config.enable_tls {
            tracing::warn!("TLS配置暂时禁用，使用明文连接");
        }
        
        // 启动服务器
        server_builder
            .add_service(service)
            .serve(self.config.bind_address)
            .await
            .map_err(|e| anyhow::anyhow!("Flight SQL服务器启动失败: {}", e))?;
        
        tracing::info!("Flight SQL服务器已停止");
        Ok(())
    }
    
    /// 获取服务器配置
    pub fn config(&self) -> &FlightSqlServerConfig {
        &self.config
    }
    
    /// 获取内核引用
    pub fn kernel(&self) -> Arc<DataGatewayKernel> {
        self.kernel.clone()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use data_gateway_kernel::DataGatewayKernel;
    
    #[test]
    fn test_flight_sql_server_config_default() {
        let config = FlightSqlServerConfig::default();
        assert_eq!(config.bind_address.to_string(), "127.0.0.1:50051");
        assert!(!config.enable_tls);
        assert!(config.tls_cert_path.is_none());
        assert!(config.tls_key_path.is_none());
        assert_eq!(config.max_concurrent_connections, 1000);
        assert_eq!(config.request_timeout_seconds, 300);
    }
    
    #[tokio::test]
    async fn test_flight_sql_server_creation() {
        let config = FlightSqlServerConfig::default();
        let kernel = Arc::new(DataGatewayKernel::new().await.unwrap());
        
        let server = FlightSqlServer::new(config, kernel.clone());
        assert_eq!(server.config().bind_address.to_string(), "127.0.0.1:50051");
        
        let kernel_ref = server.kernel();
        assert!(Arc::ptr_eq(&kernel, &kernel_ref));
    }
}
